/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */

package maxent.mr.trainer;

import java.net.URI;
import maxent.mr.io.DeltaLUtil;
import maxent.mr.io.ObjectWriter;
import maxent.mr.mapred.IISMapper;
import maxent.mr.mapred.IISReducer;
import maxent.mr.mapred.IISUpdater;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import stanford.tagger.LambdaSolveTagger;

/**
 *
 * @author hadoop
 */
public class IISRunner extends Configured implements Tool{

    private int iteration = 1;
    private LambdaSolveTagger prob;

    public IISRunner() {
    }

    public IISRunner(int iteration) {
        this.iteration = iteration;
    }

    public IISRunner(int iteration, LambdaSolveTagger prob) {
        this.prob = prob;
        this.iteration = iteration;
    }

    
    public int run(String[] args) throws Exception {

        String inputPath = "/user/hadoop/init";
        String outputPath = "/user/hadoop/iis-";
        String distCachePath = "/user/hadoop/temp";
        String distCacheFileName = "lambda-";

        // write deltaL initialization
        // double deltaL = 0.0
        DeltaLUtil dlUtil = new DeltaLUtil(prob);
        dlUtil.write(inputPath);

        int iter = 1;

        do{

            System.out.println("Iteration "+iter);

            Configuration conf = new Configuration();

            // writing LambdaSolveTrainer to temporary HDFS
            Path temp = ObjectWriter.write(conf, prob, distCachePath, (distCacheFileName+iter));
            // Register the file in the DC.  Open the local file "targets"
            DistributedCache.addCacheFile(new URI(temp+"#targets"), conf);
            DistributedCache.createSymlink(conf);


            // main mapreduce, finding newton approximation
            Job job = new Job(conf);
            job.setJarByClass(IISRunner.class);

            SequenceFileInputFormat.addInputPath(job, new Path(inputPath));
            SequenceFileOutputFormat.setOutputPath(job, new Path(outputPath + iter));

            job.setMapperClass(IISMapper.class);
            job.setReducerClass(IISReducer.class);

            job.setInputFormatClass(SequenceFileInputFormat.class);
            job.setMapOutputKeyClass(IntWritable.class);
            job.setMapOutputValueClass(DoubleWritable.class);

            job.setOutputFormatClass(SequenceFileOutputFormat.class);
            job.setOutputKeyClass(IntWritable.class);
            job.setOutputValueClass(DoubleWritable.class);
            job.waitForCompletion(true);

            
            // update the LambdaSolveTrainer
            IISUpdater updater = new IISUpdater(prob,outputPath + iter);
            prob = updater.update();

            iter++;
            
        } while(iter <= iteration);

        return 0;
        
    }

    public static void main(String[]args) throws Exception{

        System.out.println("Starting Job IIS");
        long startTime = System.currentTimeMillis();

        int iteration = 5;
        int res = ToolRunner.run(new Configuration(), new IISRunner(iteration), args);
        System.out.println("Job IIS finished in " + (System.currentTimeMillis()-startTime)/1000.0 + " seconds");
        System.exit(res);

    }

}
